package _01_quick_start._8_2_3_reactive_tweets

import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream._
import akka.stream.scaladsl._

import scala.concurrent.duration._

object Main extends App {

  implicit val system       = ActorSystem("QuickStart")
  implicit val materializer = ActorMaterializer()
  implicit val ec           = system.dispatcher

  final case class Author(handle: String)

  final case class Hashtag(name: String)

  final case class Tweet(author: Author, timestamp: Long, body: String) {
    def hashtags: Set[Hashtag] =
      body.split(" ").collect { case t if t.startsWith("#") => Hashtag(t) }.toSet
  }

  val akkaTag = Hashtag("#akka")

  /*
   * 非结构化`streaming`处理
   */

  // source
  val tweets: Source[Tweet, NotUsed] = Source(
    Set(
      Tweet(Author("Martin"), System.currentTimeMillis(), "Ordersky #akka creator"),
      Tweet(Author("Barudisu"), System.currentTimeMillis(), "#akka leaner"),
      Tweet(Author("Petter"), System.currentTimeMillis(), "Mahina #scala practise")
    ))

  // Transforming and consuming simple streams 过滤、映射
  val authors: Source[Author, NotUsed] = tweets
                                         .filter(_.hashtags.contains(akkaTag))
                                         .map(_.author)

  // Flattening sequences in streams 扁平化处理
  val hashtags: Source[Hashtag, NotUsed] = tweets.mapConcat(_.hashtags.toList)

  val done = authors.runWith(Sink foreach println)

  done.onComplete(_ => system.terminate())
}
